package com.inns.position.kafka.consumer.impl;

import com.inns.position.handler.DefaultHandler;
import com.inns.position.kafka.bean.LocationInfo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * Created by phoenix on 2018/4/7.
 */
@Component
public class LocationInfoConsumer {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Resource
    private DefaultHandler defaultHandler;

    @KafkaListener(topics = {"inns-femtocell"})
    public void listen(ConsumerRecord<?, ?> record) {
        String value = record.value().toString();
//        String format = String.format("offset=%d,topic=%s,partition=%s,key=%s,value=%s\n", record.offset(), record.topic(), record.partition(), record.key(), value);
        logger.info("ConsumerRecord is: " + value);
        LocationInfo locationInfo = new LocationInfo(value);
        try {
            defaultHandler.addToQueue(locationInfo);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

}
